/* Copyright (c) 2005 - 2016 Hewlett Packard Enterprise Development LP  -*- Java -*-*/
/* 
 * Description: Example User Defined Transform Function with parameters: Output top-k rows in each partition
 *
 * Create Date: June 1, 2013
 */

package com.vertica.JavaLibs;

import java.util.ArrayList;
import java.util.Hashtable;

import javax.naming.NamingEnumeration;
import javax.naming.directory.Attributes;
import javax.naming.directory.SearchControls;
import javax.naming.directory.SearchResult;
import javax.naming.ldap.InitialLdapContext;

import com.vertica.sdk.DestroyInvocation;
import com.vertica.sdk.PartitionReader;
import com.vertica.sdk.PartitionWriter;
import com.vertica.sdk.ServerInterface;
import com.vertica.sdk.SizedColumnTypes;
import com.vertica.sdk.TransformFunction;
import com.vertica.sdk.UdfException;

public class LdapUsers extends TransformFunction
{
   
    @Override
    public void processPartition(ServerInterface srvInterface, PartitionReader inputReader, PartitionWriter outputWriter)
                throws UdfException, DestroyInvocation
    {
    	SizedColumnTypes inTypes = inputReader.getTypeMetaData();
    	ArrayList<Integer> argCols = new ArrayList<Integer>();
    	inTypes.getArgumentColumns(argCols);
    	
    	
    	if (argCols.size() !=6
				|| !inTypes.getColumnType(argCols.get(0)).isVarchar()
				|| !inTypes.getColumnType(argCols.get(1)).isVarchar()
				|| !inTypes.getColumnType(argCols.get(2)).isInt()
    			|| !inTypes.getColumnType(argCols.get(3)).isVarchar()
				|| !inTypes.getColumnType(argCols.get(4)).isVarchar()
				|| !inTypes.getColumnType(argCols.get(5)).isVarchar()
    			) {
			throw new UdfException(0,
					"Function Expects 6 arguments(Varchar,Varchar,Integer,Varchar,Varchar, Varchar)");
		}
    	SizedColumnTypes outTypes = outputWriter.getTypeMetaData();
		ArrayList<Integer> outArgsCols = new ArrayList<Integer>();
		outTypes.getArgumentColumns(outArgsCols);
    	
    	do {
    		String domainName =inputReader.getString(argCols.get(0));
            String svpIp =inputReader.getString(argCols.get(1));
            int svpPort =Long.valueOf( inputReader.getLong(argCols.get(2)) ).intValue()  ;
            String userName =inputReader.getString(argCols.get(3));
            String password =inputReader.getString(argCols.get(4));
            String dc =inputReader.getString(argCols.get(5));
            
            
            String[] arrayOfString = { "cn", "sn", "givenName",
    				"userPrincipalName", "canonicalName", "AdsPath","sAMAccountName" };
            String str1 = "(&(objectcategory=Person)(objectclass=user))";
            
            SearchControls localSearchControls = new SearchControls();
    		localSearchControls.setReturningAttributes(arrayOfString);

    		localSearchControls.setSearchScope(2);

    		String DOMAIN_NAME = domainName.trim().toLowerCase();
    		String str2;
    		if ((dc == null) || (dc.equals("")))
    			str2 = ("." + DOMAIN_NAME).replaceAll("\\.", ",DC=").substring(1);
    		else {
    			str2 = dc;
    		}
    		String SVR_IP = svpIp.trim().toLowerCase();
    		if ((SVR_IP == null) || ("".equals(SVR_IP))) {
    			SVR_IP = DOMAIN_NAME;
    		}
    		Hashtable localHashtable = new Hashtable();
    		localHashtable.put("java.naming.factory.initial",
    				"com.sun.jndi.ldap.LdapCtxFactory");

    		localHashtable.put("java.naming.provider.url", "LDAP://" + SVR_IP + ":"
    				+ String.valueOf(svpPort));
    		localHashtable.put("java.naming.security.authentication", "simple");
    		localHashtable.put("java.naming.security.principal", userName
    				+ "@" + DOMAIN_NAME);
    		localHashtable.put("java.naming.security.credentials", password);

    		InitialLdapContext localInitialLdapContext = null;
            
    		try {
    			localInitialLdapContext = new InitialLdapContext(localHashtable,
    					null);

    			NamingEnumeration localNamingEnumeration = localInitialLdapContext
    					.search(str2, str1, localSearchControls);
    			while (localNamingEnumeration.hasMoreElements()) {
    				SearchResult localSearchResult = (SearchResult) localNamingEnumeration
    						.next();
    				if (localSearchResult != null){
    					Attributes attrs= localSearchResult.getAttributes();
    					
    					String sAMAccountName =attrs.get("sAMAccountName").getAll().next().toString();
    					String canonicalName=attrs.get("canonicalName").getAll().next().toString();
    					
    					outputWriter.setString(outArgsCols.get(0), sAMAccountName); // sAMAccountName
    					outputWriter.setString(outArgsCols.get(1), canonicalName); // sAMAccountName
    					outputWriter.next();
    				}
    			}

    		} catch (Exception e) {
    			e.printStackTrace();
    		}
            
            
            
            
            

		} while (inputReader.next());
    }
    
    
    
}
